package framework;

import java.util.Hashtable;
import java.util.Vector;

/**
 * This filter detects wild points and extrapolate the respective replacement
 * value in case of the measurement is a wild point.
 */
public class WildPointExtrapolator extends FilterFramework {

	/**
	 * The array of ids to verify and extrapolate in case of wild point
	 */
	private int[] idsToExtrapolate;
	/**
	 * The function to detect if the measurement(s) is a wild point
	 */
	private IWildPointDetector detector;

	/**
	 * The list of wild points accumulated waiting for a valid value to
	 * extrapolate.
	 */
	private Vector<Hashtable<Integer, Long>> accumulated;

	/**
	 * Represents the measurement values of the last valid frame
	 */
	private long[] lastValid;

	/**
	 * Used to detect the first valid value. To handle the special case of the
	 * first frame is a wild point
	 */
	private boolean oneValid;

	/**
	 * Constructor for a Wild Point Extrapolater filter
	 * 
	 * @param idsToExtrapolate
	 *            the ids that should be extrapolated
	 * @param detector
	 *            the function that detects if the measures are a wild point
	 */
	public WildPointExtrapolator(int[] idsToExtrapolate,
			IWildPointDetector detector) {
		super(1, 2);
		this.idsToExtrapolate = idsToExtrapolate;
		this.detector = detector;
	}

	/**
	 * 
	 * This method reads the id of a measurement from the input pipe
	 * 
	 * @return The measurement id read
	 * @throws EndOfStreamException
	 *             if the input pipe is closed or no more data in input pipe
	 */
	protected int ReadInt() throws EndOfStreamException {
		return super.ReadInt(0);
	}

	/**
	 * 
	 * This method reads the value of a measurement from the input pipe
	 * 
	 * @return The measurement value read
	 * @throws EndOfStreamException
	 *             if the input pipe is closed or no more data in input pipe
	 */
	protected long ReadLong() throws EndOfStreamException {
		return super.ReadLong(0);
	}

	/**
	 * The main method of the filter. Starts when the thread is launched
	 */
	@Override
	public void run() {
		try {
			this.accumulated = new Vector<Hashtable<Integer, Long>>();
			this.lastValid = new long[this.idsToExtrapolate.length];
			this.oneValid = false;

			while (this.ReadInt() != 0) {
				this.ReadLong();
			}

			while (true) {

				Hashtable<Integer, Long> frame = this.readFrame();

				long[] measures = new long[this.idsToExtrapolate.length];
				for (int i = 0; i < this.idsToExtrapolate.length; i++) {
					measures[i] = frame.get(this.idsToExtrapolate[i]);
				}

				boolean isWP = this.detector.detect(measures);

				if (isWP) {
					this.processWildPoint(frame);
				} else {
					this.processValidPoint(frame, measures);
				}

			} // while
		} catch (EndOfStreamException e) {
			// special case: all are wildpoints, use 0 as default value
			if (!this.oneValid) {
				for (int i = 0; i < this.idsToExtrapolate.length; i++) {
					this.lastValid[i] = 0;
				}
			}

			// there is no more frames: must send the accumulated
			this.sendAllAccumulated(this.lastValid);

			ClosePorts();
			System.out.print("\n" + this.getName() + "::WildPoint Exiting;");
		} // catch
		catch (Exception e) {
			ClosePorts();
			System.out.print("\n" + this.getName()
					+ "::ERROR::WildPoint Exiting;" + e.getMessage());
			e.printStackTrace();
		} // catch

	}

	/**
	 * 
	 * This methods reads an entire frame from the input pipe
	 * 
	 * @return a complete frame, represented by a hash of measurement ids and respective values
	 * @throws EndOfStreamException if the input pipe is closed or no more data in input pipe
	 */
	private Hashtable<Integer, Long> readFrame() throws EndOfStreamException {
		// read time
		long measure = this.ReadLong();

		// create new hash for this new frame
		Hashtable<Integer, Long> hash = new Hashtable<Integer, Long>();

		// put the value for the time in the hash
		hash.put(Constants.TIME_ID, measure);

		try {
			for (int id = this.ReadInt(); id != 0; id = this.ReadInt()) {
				measure = this.ReadLong();
				// put the value read in the hash
				hash.put(id, measure);
			}
		} catch (EndOfStreamException e) {
			return hash;
		}

		return hash;
	}

	/**
	 * This method processes a wild point frame
	 * 
	 * @param the frame the frame to be processed
	 */
	private void processWildPoint(Hashtable<Integer, Long> frame) {
		this.accumulated.add(frame);
		this.sendToWPSink(frame);
	}

	/**
	 * This method processes a valid frame
	 * @param frame the frame the frame to be processed
	 * @param newValid The most recent value valued to be used in the extrapolation.
	 */
	private void processValidPoint(Hashtable<Integer, Long> frame,
			long[] newValid) {

		// special case for the first valid point
		if (!this.oneValid) {
			this.oneValid = true;
			this.lastValid = newValid;
		}

		// calculate the extrapolated values
		long[] extrapolValues = new long[this.idsToExtrapolate.length];
		for (int i = 0; i < this.idsToExtrapolate.length; i++) {
			double lv = Double.longBitsToDouble(this.lastValid[i]);
			double nv = Double.longBitsToDouble(newValid[i]);
			double extrapol = (lv + nv) / 2.0;
			extrapolValues[i] = Double.doubleToLongBits(extrapol);
		}

		// set the last valid as this one
		this.lastValid = newValid;

		this.sendAllAccumulated(extrapolValues);

		// send the current frame
		this.sendFrame(frame);

		// clean the accumulated vector
		this.accumulated = new Vector<Hashtable<Integer, Long>>();
	}

	/**
	 * 
	 * This methods sends all the previous accumulated values 
	 * (wild points waiting for a valid value) to the output pipe
	 * 
	 * @param extrapolValues the new values (extrapoled) for the each one of the respective measurements
	 */
	private void sendAllAccumulated(long[] extrapolValues) {
		for (Hashtable<Integer, Long> frameWP : this.accumulated) {
			Long currentMask = frameWP.get(Constants.MASK_ID);
			long mask = 0;
			if (currentMask != null) {
				mask = currentMask;
			}
			// change the WP values to the extrapolated ones
			for (int i = 0; i < this.idsToExtrapolate.length; i++) {
				frameWP.put(this.idsToExtrapolate[i], extrapolValues[i]);
				mask = (1 << this.idsToExtrapolate[i] | mask);
			}
			frameWP.put(Constants.MASK_ID, mask);

			// send the extrapolated frame
			this.sendFrame(frameWP);
		}
	}

	/**
	 * This methods writes the frame (the list of measurement ids and respective values)
	 *  to the output pipe
	 * @param frame the frame to be written in the output pipe
	 */
	private void sendFrame(Hashtable<Integer, Long> frame) {
		// send the id 0 -> timestamp
		this.writeInt(Constants.TIME_ID, 0);
		this.writeLong(frame.remove(Constants.TIME_ID), 0);

		// send the other ids
		for (int id : frame.keySet()) {
			this.writeInt(id, 0);
			this.writeLong(frame.get(id), 0);
		}
	}

	/**
	 * This method writes the wild point to the respective output pipe 
	 * (the one connected to the wild point sink)
	 * @param frame the frame to be written in the output pipe
	 */
	private void sendToWPSink(Hashtable<Integer, Long> frame) {
		this.writeInt(Constants.TIME_ID, 1);
		this.writeLong(frame.get(Constants.TIME_ID), 1);

		for (int id : this.idsToExtrapolate) {
			this.writeInt(id, 1);
			this.writeLong(frame.get(id), 1);
		}
	}
}
